/************************************************************************
Copyright 2003, 2004, 2005, 2006, 2007, 2008 Grid Technology Research Center,
National Institute of Advanced Industrial Science and Technology.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
************************************************************************/

package jp.aist.gtrc.plus.scheduler.job;

import org.apache.log4j.Logger;

import java.util.Collection;
import java.util.Map;

import jp.aist.gtrc.plus.scheduler.SchedulerOptions;
import jp.aist.gtrc.plus.scheduler.status.JobStatus;
import jp.aist.gtrc.plus.scheduler.sys.SchedulerException;
import jp.aist.gtrc.plus.scheduler.util.CollectionUtil;
import jp.aist.gtrc.plus.scheduler.util.Sorter;
import jp.aist.gtrc.plus.scheduler.util.SortKey;


public class SortScheduler  extends Scheduler {

        protected static final Logger logger = Logger.getLogger(SortScheduler.class);

	private final Sorter<JobStatus> sorter;

	private static final String TITLE_JOB_SORTKEY = "JobSortKey";
	private static final String DELIM_KEYS = ",\\s*";

	public SortScheduler() {

		/*
		 * We must give higher priority to reserved jobs
		 * than non-reserved jobs always.
		 * And it must be first (prime) sort key, so add it here.
		 */
		sorter = new Sorter<JobStatus>();
		addJobSortKey(JobSortKey.ReservePrior);

	}

	final void addJobSortKey(JobSortKey key) {

		sorter.addSortKey(key);
		logger.info("Added JobSortKey: " + key);

	}

        final void addJobSortKey(SortKey<JobStatus> key) {
                sorter.addSortKey(key);
		logger.info("Added SortKey: " + key);
        }

	private void setJobSorter(Map<String, String> confs) {

		assert(confs != null);

		String jobKeyConf = confs.get(TITLE_JOB_SORTKEY);
		if (jobKeyConf == null)
			return;

		for (String keyName : jobKeyConf.split(DELIM_KEYS)) {
			JobSortKey key = JobSortKey.valueOf(keyName);
			if (key == null) {
				logger.warn("Unknown job sort key: "
						+ keyName + ", ignored");
				continue;
			}
			if (sorter.containsKey(key)) {
				logger.warn("Duplicate job sort key: "
						+ keyName + ", refers first one");
				continue;
			}

			if (key.needOrders()) {
				/*
				 * Set Job Owner's (group/host/domain) Orders
				 *	If keyName is "OwnersName", key.getConfigTitle() is
				 *	"OwnersOrder", and so on.
				 */
				String names = confs.get(key.getConfigTitle());
				if (names == null){
					logger.warn(keyName + " needs to specify"
							+ key.getConfigTitle());
					continue;
				}
				String[] orders = names.split(DELIM_KEYS);
				if (orders.length == 0) {
					logger.warn(key.getConfigTitle()
							+ " needs to specify order names");
					continue;
				}
				key.setOrders(orders);
			}

			// use addFirst() to sort by *least* important key at *first*,
			// sort by *most* important key at *last*.
			// See sortJobs()
			addJobSortKey(key);
		}
		if (sorter.size() == 0){
			logger.warn("No valid job sort key. Run jobs as Fifo.");
		}

	}


	public void init(SchedulerOptions options) throws SchedulerException {

		Map<String, String> confs = options.getConfig();
		if (confs != null) {
			setJobSorter(confs);
		} else {
			// config file doesn't exist.
			// We sort jobs same as default policy of Torque's fifo scheduler.
			logger.info("config file not specified, sort jobs as Torque does.");
			addJobSortKey(JobSortKey.QueuePriority);
			addJobSortKey(JobSortKey.ByQueue);
			addJobSortKey(JobSortKey.JobPriority);
		}

		/*
		 * If there are free(idle) execute nodes, jobs are started
		 * as soon as submitted regardless sort orders specified above.
		 *
		 * If all nodes are busy at submit, submitted jobs are queued.
		 * After some running job is exited, regardless order in queue,
		 * a job with highest priority in queue is started.
		 *
		 * start		owner's			job's requested
		 * order	group	name	priority	time
		 * ------------------------------------------
		 * 1		prof.	knuth	 0			60min
		 * 2		doct.	nakada	 100		10min
		 * 3		doct.	nakada	 0			60min
		 * 4		doct.	takefusa 500		5min
		 * 5		users	someoneX 1000		15min
		 * 6		users	someoneY 0			15min
		 * ....
		 *
		 * Job owner's group is a primary group of owner
		 * which specified in /etc/passwd & /etc/group.
		 *
		 * Job priority is specified by 'qsub -p n'.
		 * Larger n	means higher priority (0 is default, 1023 is highest).
		 *
		 * Professor knuth is *not* in OwnersName list above,
		 * but his job is started at first because he is in professors group
		 * which has highest priority and owner's group is primary sort key.
		 * Both nakada & takefusa are in same doctors group,
		 * but nakada has higher priority than takefusa.
		 * nakada's jobs are started in job priority order.
		 * someoneX in users group requested highest (=1000) priority,
		 * but it runs after professors&doctors jobs are started.
		 */

	}

	public JobStatus selectJob(Collection<JobStatus> jobs) {

		if (CollectionUtil.isEmpty(jobs))
			return null;

		JobStatus[] jobArray = jobs.toArray(new JobStatus[jobs.size()]);
		sorter.sort(jobArray);
                logger.debug("Job " + jobArray[0].getJobID() + " selected");
		return jobArray[0];

	}

}
